package guda.mrpc.core.transport;

import guda.mrpc.core.api.Request;
import guda.mrpc.core.client.ConnectionFactory;
import guda.mrpc.core.client.RpcFuture;
import guda.mrpc.core.registry.ServiceMeta;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by well on 17/5/23.
 */
public abstract  class ClientHandler<T> extends ChannelInboundHandlerAdapter {

    public static final Logger log = LoggerFactory.getLogger(ClientHandler.class);


    protected Map<String, RpcFuture> mapCallBack = new ConcurrentHashMap<>();

    protected volatile Channel channel;

    protected SocketAddress remoteSocketAddress;

    protected String serverAddress;

    private SocketAddress localAddress;
    private ServiceMeta serviceMeta;

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
        this.localAddress=ctx.channel().localAddress();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remoteSocketAddress = this.channel.remoteAddress();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        ConnectionFactory.instance().remove(this);
        log.warn("channel inactive,remote address:" + remoteSocketAddress + ",local address:" + localAddress);
        ConnectionFactory.instance().connect(serviceMeta);

    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("", cause);
        ctx.close();
    }

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

    public abstract RpcFuture sendRequest(Request request);

    public Channel getChannel() {
        return channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    public ServiceMeta getServiceMeta() {
        return serviceMeta;
    }

    public void setServiceMeta(ServiceMeta serviceMeta) {
        this.serviceMeta = serviceMeta;
    }

    public String getServerAddress() {
        return serverAddress;
    }

    public void setServerAddress(String serverAddress) {
        this.serverAddress = serverAddress;
    }

    public String toString(){
        StringBuilder buf = new StringBuilder();
        buf.append("client local address:").append(localAddress).append(",remote address:")
                .append(remoteSocketAddress).append(",service:").append(serviceMeta);
        return buf.toString();
    }


}
